package controller;

import operators.UV_ProcessWindowFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import pojo.UniqueView;
import pojo.UserBehavior;

import java.net.URL;
import java.util.HashSet;
import java.util.Random;

public class UniqueVisitor_Modify_01 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(4);

        URL resource = UniqueVisitor.class.getResource("/UserBehavior.csv");
        DataStream<UserBehavior> fileDataStream = env.readTextFile(resource.getPath())
                .map((MapFunction<String, UserBehavior>) line -> {
                    String[] fields = line.split(",");
                    return new UserBehavior(Long.valueOf(fields[0]), Long.valueOf(fields[1]), Integer.valueOf(fields[2]), fields[3], Long.valueOf(fields[4]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<UserBehavior>noWatermarks().withTimestampAssigner((SerializableTimestampAssigner<UserBehavior>) (element, recordTimestamp) -> element.getTimestamp() * 1000L))
                .filter( value -> "pv".equals(value.getBehavior()));

//        KeyedStream<UniqueView, Long> keyedStream =
                fileDataStream.map((MapFunction<UserBehavior, Tuple2<Integer,Long>>) line -> new Tuple2<>(new Random().nextInt(10),line.getUserId()))
                .returns(TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(Integer.class, Long.class))
                .keyBy( data -> data.f0 )
                .window(TumblingEventTimeWindows.of(Time.hours(1)))
                .apply(new WindowFunction<Tuple2<Integer, Long>, UniqueView, Integer, TimeWindow>() {
                    @Override
                    public void apply(Integer integer, TimeWindow window, Iterable<Tuple2<Integer, Long>> input, Collector<UniqueView> out) throws Exception {
                        input.forEach(tuple2-> out.collect(new UniqueView(tuple2.f1.toString(), window.getEnd(), 1)));
                    }
                })
                .keyBy(UniqueView::getTimestamp)
                .process(new KeyedProcessFunction<Long, UniqueView, UniqueView>() {
                    private ListState<UniqueView> uvListState;

                    @Override
                    public void open(Configuration parameters) {
                        uvListState = getRuntimeContext().getListState(
                                new ListStateDescriptor<>("uv", UniqueView.class)
                        );
                    }

                    @Override
                    public void processElement(UniqueView value, Context ctx, Collector<UniqueView> out) throws Exception {
                        ctx.timerService().registerEventTimeTimer(value.getTimestamp() + 1);
                        uvListState.add(value);
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<UniqueView> out) throws Exception {
                        HashSet<Long> userSet = new HashSet<>();
                        for (UniqueView uniqueView : uvListState.get()) {
                            userSet.add(Long.valueOf(uniqueView.getType()));
                        }

                        out.collect(new UniqueView("uv", timestamp - 1, userSet.size()));
                        uvListState.clear();
                    }
                })
                .print();

        env.execute();
    }
}
